package com.example.service.other;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;

import static com.example.entity.constant.redis.SeckillConstant.SSE_EMITTERS_KEY;

@Service
public class SseService {

    // 存储令牌与SseEmitter的映射关系
    private final Map<String, SseEmitter> emitters = new ConcurrentHashMap<>();

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    @Qualifier("sseExecutor")
    private Executor executor;

    // SSE连接超时时间（300_000L / 60_000L = 5分钟）
    private static final Long SSE_TIMEOUT = 300_000L;

    /**
     * 创建SSE连接
     * @param token 秒杀令牌
     * @return SseEmitter对象
     */
    public SseEmitter createSseConnection(String token) {
        // 创建SSE发射器，设置超时时间
        SseEmitter emitter = new SseEmitter(SSE_TIMEOUT);

        // 存储令牌与发射器的映射关系
        emitters.put(token, emitter);
        // 同时存储到Redis，便于分布式环境下使用
        redisTemplate.opsForHash().put(SSE_EMITTERS_KEY, token, "active");

        // 连接关闭时的回调
        emitter.onCompletion(() -> {
            emitters.remove(token);
            redisTemplate.opsForHash().delete(SSE_EMITTERS_KEY, token);
        });

        // 连接超时的回调
        emitter.onTimeout(() -> {
            emitters.remove(token);
            redisTemplate.opsForHash().delete(SSE_EMITTERS_KEY, token);
            try {
                emitter.send(SseEmitter.event().name("timeout").data("连接超时"));
                emitter.complete();
            } catch (IOException e) {
                // 忽略异常
            }
        });

        // 连接出错的回调
        emitter.onError((e) -> {
            emitters.remove(token);
            redisTemplate.opsForHash().delete(SSE_EMITTERS_KEY, token);
        });

        return emitter;
    }

    /**
     * 推送秒杀结果给客户端
     * @param token 秒杀令牌
     * @param result 秒杀结果
     */
    @Async("sseExecutor")
    public void pushSeckillResult(String token, String result) {
        SseEmitter emitter = emitters.get(token);
        if (emitter != null) {
            try {
                // 发送结果
                emitter.send(SseEmitter.event()
                        .name("seckillResult")
                        .data(result));
                // 完成后关闭连接
                emitter.complete();
            } catch (IOException e) {
                // 发送失败，可能是客户端已断开连接
                emitter.completeWithError(e);
            } finally {
                // 清理资源
                emitters.remove(token);
//                redisTemplate.opsForHash().delete(SSE_EMITTERS_KEY, token); //不用主动删除，后续用户一天内进来时直接获取
            }
        } else {
            // 如果本地没有找到emitter，可能是在其他节点，这里可以根据实际分布式架构进行处理（不实际）
            // 例如：发送到Redis Pub/Sub，让所有节点尝试推送
//            redisTemplate.convertAndSend("seckill:result:pubsub", token + ":" + result);

            System.out.println("sse的token不存在，用户主动退出了");
        }
    }
}
